package com.gin.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object L03_Sort {

  /*
  面向数据集操作
  1.带函数的非聚合: map, flatmap
  2.单元素: union, cartesian 没有函数计算
  3.kv元素: cogroup, join 没有函数计算
  4.排序
  5.聚合计算: reduceByKey 有函数 combinerByKey
  * */
  def main(args: Array[String]): Unit = {
    //加载配置 获取spark上下文对象
    val conf = new SparkConf().setMaster("local").setAppName("L01")
    val sc = new SparkContext(conf)

    /*
    * 数据源 PV-用户浏览一次记一次  UV-用户浏览(用户维度)去重数
    * 需求:根据数据计算各网站的PV,UV; 同时只显示top5
    * */

    val fileRDD: RDD[String] = sc.textFile("scala/data/pvuvData")

    //统计 网站分组, 计数
    //43.169.217.152	河北	2018-11-12	1542011088714	3292380437528494072	www.dangdang.com	Login
    //取下标为5的数据(www.dangdang.com), 并将数据映射为 (www.dangdang.com, 1)
    val pair: RDD[(String, Int)] = fileRDD.map(line => (line.split("\t")(5), 1))
    val reduce: RDD[(String, Int)] = pair.reduceByKey(_ + _)
    val map: RDD[(Int, String)] = reduce.map(_.swap)
    //false 倒序排序
    val sorted: RDD[(Int, String)] = map.sortByKey(false)
    val res: RDD[(String, Int)] = sorted.map(_.swap)
    val pv: Array[(String, Int)] = res.take(5)
    pv.foreach(println)

    println("-------- pv --------")
    println()



    //统计 网站+IP(去重), 按网站分组, 计数
    val keys: RDD[(String, String)] = fileRDD.map(
      line => {
        val strs: Array[String] = line.split("\t")
        new Tuple2(strs(5), strs(0))
      }
    )
    //针对二元组里两个元素进行去重
    val key: RDD[(String, String)] = keys.distinct()
    //取网站进行映射, 以备计数统计
    val mapUV: RDD[(String, Int)] = key.map(k => (k._1, 1))
    val uvReduce: RDD[(String, Int)] = mapUV.reduceByKey(_ + _)
    //根据二元组的第二个元素进行倒序排序
    val uvSorted: RDD[(String, Int)] = uvReduce.sortBy(_._2, false)
    val uv: Array[(String, Int)] = uvSorted.take(5)
    uv.foreach(println)

    println("-------- uv --------")
    println()




  }

}
